/*
 * Created on 2007-11-2
 *
 */
package jacky.lanlan.song.test.common;

import java.lang.reflect.*;
import java.util.*;
import java.util.concurrent.*;

/**
 * 并发执行相关的实用工具类。
 * @author Jacky.Song
 */
public abstract class ConcurrentRunUtils {
	
	private static final int DEFAULT_THREAD_COUNT=10;
	
	//线程池
	private static ScheduledThreadPoolExecutor pool=new ScheduledThreadPoolExecutor(DEFAULT_THREAD_COUNT);
	
	static {
		pool.prestartAllCoreThreads();
	}
	
	private static void resetPool(int threadCount) {
		pool.setMaximumPoolSize(threadCount<DEFAULT_THREAD_COUNT?DEFAULT_THREAD_COUNT:threadCount);
		pool.setKeepAliveTime(0, TimeUnit.MILLISECONDS);
	}

	private static void mustPositive(int... values) {
		for (int value : values) {
			Assert.isTrue(value>0, "参数必须大于0");
		}
	}
	
	/**
	 * 并发、反复执行指定任务，用于测试某个类的线程安全性或某个方法的并发性能。
	 * <p/>
	 * 任务被每个线程反复执行，线程均分执行数量，每10ms执行一次。
	 * <p/>
	 * <b style="color:red">注意，这个方法将阻塞调用线程，直到所有任务执行完毕才返回。</b>
	 * @param task 需要并发执行的任务
	 * @param threadCount 执行线程数
	 * @param times 执行任务的次数
	 */
	public static void concurrentRepetitiveExecution(final Runnable task,int threadCount,int times) {
		mustPositive(threadCount,times);
		
		if(threadCount<=0) throw new IllegalArgumentException("threadCount 必须为正数");
		if(times<=0) throw new IllegalArgumentException("times 必须为正数");
		
		final ScheduledFuture<?> [] futures=new ScheduledFuture<?> [threadCount];

		resetPool(threadCount);
		
		final CountDownLatch startSignal = new CountDownLatch(1);
		
		int intervalPerThread=10;//线程每次执行的间隔时间(毫秒)
		for (int i = 0; i < futures.length; i++) {
			futures[i]=pool.scheduleAtFixedRate(new Runnable() {
				public void run() {
					try {
						startSignal.await();//等待开始信号
					}
					catch (InterruptedException e) {}
					task.run();
				}
			}, 0, intervalPerThread, TimeUnit.MILLISECONDS);
		}
		
		startSignal.countDown();//发出开始信号
		
		//一个线程的运行时间(毫秒)
		int threadRunningTime=times/threadCount*intervalPerThread;
		//所以threadCount个线程的总共执行次数为：总线程数×每个线程执行的次数，而每个线程执行的次数=线程运行时间/每次执行的间隔时间
		//即：次数= threadCount * (times/threadCount*intervalPerThread_Task) * intervalPerThread_Task = times次
		
		pool.schedule(new Runnable() {//停止任务
			public void run() {
				for (ScheduledFuture<?> future : futures) {
					future.cancel(true);
				}
			}
		},threadRunningTime,TimeUnit.MILLISECONDS);
		
		//检测任务是否都被手动取消
		waitUntilAllPass(Arrays.asList(futures), new Judgment<ScheduledFuture<?>>() {
			public boolean canPass(ScheduledFuture<?> arg) {
				return arg.isCancelled();
			}
		},pool);

	}

	/**
	 * 启动5个线程并发、反复执行指定的任务，线程均分执行数量，每10ms执行一次。
	 * @param task 需要并发执行的任务
	 * @param times 执行任务的次数
	 * @see #concurrentRepetitiveExecution(Runnable, int, int)
	 */
	public static void concurrentRepetitiveExecution(Runnable task,int times) {
		concurrentRepetitiveExecution(task,5,times);
	}
	
	/**
	 * 启动5个线程并发执行指定任务100次。每个线程执行20次任务，每10ms执行一次。
	 * @param task 需要并发执行的任务
	 * @see #concurrentRepetitiveExecution(Runnable, int, int)
	 */
	public static void concurrentRepetitiveExecution(Runnable task) {
		concurrentRepetitiveExecution(task,5,100);
	}
	
	/**
	 * 同时并发执行指定任务，用于测试某个类的线程安全性或某个方法的并发性能。
	 * <p/>
	 * 每个线程执行一次任务，尽可能的同时并发执行。(之所以说尽可能，是因为想要这些线程精确的同时执行几乎不可能)
	 * <p/>
	 * <b style="color:red">注意，这个方法将阻塞调用线程，直到所有任务执行完毕才返回。</b>
	 * @param <T>
	 * @param task 需要并发执行的任务
	 * @param times 执行任务的次数
	 * @return 任务执行后的结果
	 */
	public static <T> T[] concurrentExecution(Callable<T> task,int times) {
		mustPositive(times);
		
		Object [] results=new Object[times];
		resetPool(times);
		Future<T> [] futures=executeTogather(task, times);
		try {
			for(int i=0;i<times;i++) {
				results[i]=futures[i].get();
			}
		}
		catch (Exception e) {}
		
		//下面把Object[]转换为泛型数组，注意只能这样转，不能直接强转:(T[])results
		T[] arr=(T[]) Array.newInstance(results[0].getClass(), times);//先重新创建一个数组，然后强转为泛型数组
		System.arraycopy(results, 0, arr, 0, times);//再把Object[]里面的拷贝到泛型数组
		return arr;
	}

	/**
	 * 同时并发执行指定任务，用于测试某个类的线程安全性或某个方法的并发性能。
	 * <p/>
	 * 每个线程执行一次任务，尽可能的同时并发执行。(之所以说尽可能，是因为想要这些线程精确的同时执行几乎不可能)
	 * <p/>
	 * <b style="color:red">注意，这个方法将阻塞调用线程，直到所有任务执行完毕才返回。</b>
	 * @param task 需要并发执行的任务
	 * @param times 执行任务的次数
	 */
	public static void concurrentExecution(final Runnable task,int times) {
		mustPositive(times);
		
		Future<?> [] futures=executeTogather(new Callable<Object>() {//一个Callable的Adapter
			public Object call() throws Exception {
				task.run();
				return null;
			}
		}, times);
		
		resetPool(times);
		
		//检测任务是否都已完成
		waitUntilAllPass(Arrays.asList(futures), new Judgment<Future<?>>() {
			public boolean canPass(Future<?> arg) {
				return arg.isDone();
			}
		},pool);
	}

	//尽可能的让pool里的线程同时执行task，一共times次，目的是创造一个并发的环境
	private static <T> Future<T> [] executeTogather(final Callable<T> task,int times){
		Future<T> [] futures=new Future[times];
		final CountDownLatch startSignal = new CountDownLatch(1);
		
		final Random ran = new Random();
		for(int i=0;i<times;i++) {
			futures[i]=pool.submit(new Callable<T>() {
				public T call() throws Exception {
					startSignal.await();
					Thread.sleep(ran.nextInt(30));// 希望可以创造机会让线程同时醒来
					T reValue=task.call();
					return reValue;
				}
			});
		}
		
		startSignal.countDown();//发出开始执行信号
		
		return futures;
	}

	private static <T> void waitUntilAllPass(Collection<T> items,final Judgment<T> condition,
			final Executor executor){
    final CountDownLatch doneSignal = new CountDownLatch(items.size());
    for (final T item:items) {
    	//提交给线程池来执行判断
    	executor.execute(new Runnable() {
				public void run() {
					boolean pass= condition.canPass(item);//执行任务
					if(!pass) {
						executor.execute(this);//如果没有通过，重新把自己放回池中
					}
					else {
						doneSignal.countDown();//如果通过，发出完成信号
					}
				}
			});
    }
    try {
			doneSignal.await();//阻塞，直到所有任务完成
		}
		catch (InterruptedException e) {
			throw new RuntimeException(e);
		}
	}

	private static interface Judgment<T>{
		boolean canPass(T arg);
	}
}
